package com.bootdo.common.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.bootdo.common.domain.MyTuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisZSetCommands;
import org.springframework.data.redis.core.*;
import org.springframework.stereotype.Component;
import org.xerial.snappy.Snappy;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

/**
 * Redis 工具类
 *
 * @author jinse
 */
@Component
public class RedisUtils {

    private final Logger logger = LoggerFactory.getLogger(RedisUtils.class);

    @Autowired
    private StringRedisTemplate redisTemplate;

    public BigInteger getBigInteger(String key) {
        return redisTemplate.execute((RedisCallback<BigInteger>) connection -> {
            byte[] bvalue = connection.get(key.getBytes());
            if (bvalue == null) {
                return new BigInteger("0");
            } else {
                try {
                    byte[] uncompressed = Snappy.uncompress(bvalue);
                    return new BigInteger(redisTemplate.getStringSerializer().deserialize(uncompressed));
                } catch (IOException e) {
                    logger.error("Redis getBigInteger error:", e);
                }
                return new BigInteger("0");
            }
        });
    }

    public String getString(String key) {
        return redisTemplate.execute((RedisCallback<String>) connection -> {
            byte[] bvalue = connection.get(key.getBytes());
            if (bvalue == null) {
                return null;
            } else {
                try {
                    byte[] uncompressed = Snappy.uncompress(bvalue);
                    return redisTemplate.getStringSerializer().deserialize(uncompressed);
                } catch (IOException e) {
                    logger.error("Redis getString error:", e);
                }
                return null;
            }
        });
    }

    public void expire(final String key, final Long seconds) {
        this.redisTemplate.execute((RedisCallback<Object>) connection -> {
            connection.expire(key.getBytes(), seconds);
            return 1L;
        });
    }

    public void set(final String key, final String value) {
        this.redisTemplate.execute((RedisCallback<Object>) connection -> {
            // compress data
            try {
                byte[] compressed = Snappy.compress(value);
                connection.set(key.getBytes(), compressed);
            } catch (IOException e) {
                logger.error("Redis set error:", e);
            }
            return 1L;
        });
    }

    public void set(final String key, final Object value) {
        this.redisTemplate.execute((RedisCallback<Object>) connection -> {
            // compress data
            String json = JSON.toJSONString(value, SerializerFeature.BrowserCompatible);
            try {
                byte[] compressed = Snappy.compress(json);
                connection.set(key.getBytes(), compressed);
            } catch (IOException e) {
                logger.error("Redis set error:", e);
            }
            return 1L;
        });
    }

    public <T> T get(final String key, final Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<T>) conn -> {
            byte[] res = conn.get(key.getBytes());
            if (res == null) {
                return null;
            } else {
                try {
                    byte[] uncompressed = Snappy.uncompress(res);
                    String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                    return JSON.parseObject(json, clazz);
                } catch (IOException e) {
                    logger.error("Redis get error:", e);
                }
                return null;
            }
        });
    }

    public void hSet(final String key, final String field, final Object value) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> {
            // compress data
            String json = JSON.toJSONString(value, SerializerFeature.BrowserCompatible);
            try {
                byte[] compressed = Snappy.compress(json);
                conn.hSet(key.getBytes(), field.getBytes(), compressed);
            } catch (IOException e) {
                logger.error("Redis hSet error:", e);
            }
            return null;
        });
    }

    public <T> T hGet(final String key, final String field, final Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<T>) conn -> {
            byte[] bvalue = conn.hGet(key.getBytes(), field.getBytes());
            if (null == bvalue) {
                return null;
            } else {
                try {
                    byte[] uncompressed = Snappy.uncompress(bvalue);
                    String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                    return JSON.parseObject(json, clazz);
                } catch (IOException e) {
                    logger.error("Redis hGet error:", e);
                }
                return null;
            }
        });
    }

    public void rPush(final String key, final Object value) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> {
            try {
                String json = JSON.toJSONString(value, SerializerFeature.BrowserCompatible);
                // compress data
                byte[] compressed = Snappy.compress(json);

                conn.rPush(key.getBytes(), compressed);
            } catch (IOException e) {
                logger.error("Redis rPush error:", e);
            }
            return null;
        });
    }

    public <T> T lPop(final String key, final Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<T>) conn -> {
            byte[] res = conn.lPop(key.getBytes());
            if (res == null) {
                return null;
            }
            try {
                byte[] uncompressed = Snappy.uncompress(res);
                String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                return JSON.parseObject(json, clazz);
            } catch (IOException e) {
                logger.error("Redis lPop error:", e);
            }
            return null;
        });
    }

    public Long lLen(final String key) {
        return this.redisTemplate.execute((RedisCallback<Long>) conn -> conn.lLen(key.getBytes()));
    }

    public void del(final String key) {
        this.redisTemplate.execute((RedisCallback<Long>) conn -> conn.del(key.getBytes()));
    }


    public void hDel(final String key, final String field) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> conn.hDel(key.getBytes(), field.getBytes()));
    }

    public <T> Cursor<Map.Entry<String, T>> hScan(String key, ScanOptions options, Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<Cursor<Map.Entry<String, T>>>) conn -> new ConvertingCursor<>(
                conn.hScan(key.getBytes(), options),
                source -> new Map.Entry<String, T>() {
                    @Override
                    public String getKey() {
                        return redisTemplate.getStringSerializer().deserialize(source.getKey());
                    }

                    @Override
                    public T getValue() {
                        try {
                            byte[] uncompressed = Snappy.uncompress(source.getValue());
                            String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                            return JSON.parseObject(json, clazz);
                        } catch (IOException e) {
                            logger.error("Redis hscan error:", e);
                        }
                        return null;
                    }

                    @Override
                    public T setValue(T value) {
                        throw new UnsupportedOperationException("Cannot set value for entry in cursor");
                    }
                }));
    }


    public void zAdd(final String key, final BigDecimal score, final String member) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> conn.zAdd(key.getBytes(), score.doubleValue(), member.getBytes()));
    }

    public BigDecimal zScore(final String key, final String member) {
        return this.redisTemplate.execute((RedisCallback<BigDecimal>) conn -> {
            Double score = conn.zScore(key.getBytes(), member.getBytes());
            if (score == null) {
                return new BigDecimal(0);
            }
            return new BigDecimal(score);
        });
    }

    public void zRem(final String key, final String member) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> conn.zRem(key.getBytes(), member.getBytes()));
    }

    public Set<MyTuple> zRangeByScoreWithScores(final String key, final BigDecimal min) {
        return this.redisTemplate.execute((RedisCallback<Set<MyTuple>>) conn -> {
            Set<RedisZSetCommands.Tuple> set = conn.zRangeByScoreWithScores(key.getBytes(),
                    RedisZSetCommands.Range.range().gte(min.doubleValue()));
            Set<MyTuple> objectSet = new HashSet<>();
            for (RedisZSetCommands.Tuple res : set) {
                if (res == null) {
                    return null;
                } else {
                    objectSet.add(new MyTuple(new String(res.getValue()), res.getScore()));
                }
            }
            return objectSet;
        });
    }

    public Cursor<Map.Entry<String, BigDecimal>> zScan(final String key, ScanOptions options) {
        return this.redisTemplate.execute((RedisCallback<Cursor<Map.Entry<String, BigDecimal>>>) conn -> new ConvertingCursor<>(
                conn.zScan(key.getBytes(), options),
                source -> new Map.Entry<String, BigDecimal>() {
                    @Override
                    public String getKey() {
                        return redisTemplate.getStringSerializer().deserialize(source.getValue());
                    }

                    @Override
                    public BigDecimal getValue() {
                        return new BigDecimal(source.getScore());
                    }

                    @Override
                    public BigDecimal setValue(BigDecimal value) {
                        throw new UnsupportedOperationException("Cannot set value for entry in cursor");
                    }
                }));
    }

    public void sAdd(final String key, final Object member) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> {
                    String json = JSON.toJSONString(member, SerializerFeature.BrowserCompatible);
                    try {
                        byte[] compressed = Snappy.compress(json);
                        conn.sAdd(key.getBytes(), compressed);
                    } catch (IOException e) {
                        logger.error("Redis sAdd error:", e);
                    }
                    return 1L;
                }

        );
    }

    public <T> T sPop(final String key, final Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<T>) conn -> {
            byte[] res = conn.sPop(key.getBytes());
            if (res == null) {
                return null;
            } else {
                try {
                    byte[] uncompressed = Snappy.uncompress(res);
                    String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                    return JSON.parseObject(json, clazz);
                } catch (IOException e) {
                    logger.error("Redis sPop error:", e);
                }
                return null;
            }
        });
    }

    public <T> T sRandMember(final String key, final Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<T>) conn -> {
            byte[] res = conn.sRandMember(key.getBytes());
            if (res == null) {
                return null;
            } else {
                try {
                    byte[] uncompressed = Snappy.uncompress(res);
                    String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                    return JSON.parseObject(json, clazz);
                } catch (IOException e) {
                    logger.error("Redis sRandMember error:", e);
                }
                return null;
            }
        });
    }

    public void sRem(final String key, final Object member) {
        this.redisTemplate.execute((RedisCallback<Object>) conn -> {
            String json = JSON.toJSONString(member, SerializerFeature.BrowserCompatible);
            try {
                byte[] compressed = Snappy.compress(json);
                conn.sRem(key.getBytes(), compressed);
            } catch (IOException e) {
                logger.error("Redis sRem error:", e);
            }
            return null;
        });

    }

    public <T> Set<T> sMembers(final String key, final Class<T> clazz) {
        return this.redisTemplate.execute((RedisCallback<Set<T>>) conn -> {
            Set<byte[]> set = conn.sMembers(key.getBytes());
            Set<T> objectSet = new HashSet<>();
            Iterator<byte[]> it = set.iterator();
            while (it.hasNext()) {
                byte[] res = it.next();
                if (res == null) {
                    return objectSet;
                } else {
                    try {
                        byte[] uncompressed = Snappy.uncompress(res);
                        String json = redisTemplate.getStringSerializer().deserialize(uncompressed);
                        objectSet.add(JSON.parseObject(json, clazz));
                    } catch (IOException e) {
                        logger.error("Redis sMembers error:", e);
                    }
                }
            }
            return objectSet;
        });

    }

}
